package com.deepglint.tableapi.udf;

import com.deepglint.beans.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.types.Row;

/**
 * @author mj
 * @version 1.0
 * @date 2021-11-28 18:31
 * 自定义sql的处理函数
 */
public class UDFTest_ScalarFunction {
    public static void main(String[] args) throws Exception {
        // 1.创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 2.读取数据
        String path = "C:\\Users\\马军\\Desktop\\Idea-workspace\\flink\\src\\main\\resources\\source.txt";
        DataStreamSource<String> sourceStream = env.readTextFile(path);

        // 3.转换pojo
        SingleOutputStreamOperator<SensorReading> mapStream = sourceStream.map(line -> {
            String[] split = line.split(",");
            return new SensorReading(split[0], split[1], new Long(split[2]), new Double(split[3]));
        });

        // 4.将流转换为表
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        Table sensorTable = tableEnv.fromDataStream(mapStream, "id,timestamp as ts,temperature as temp");

        // 5. 自定义的标量函数，实现id的hash值
        // 5.1 table api
        HashCode hashCode = new HashCode(12);
        // 需要在环境中注册UDF
        tableEnv.registerFunction("hashcode",hashCode);
        Table resultTable = sensorTable.select("id,ts,hashcode(id)");

        // 5.2 sql
        tableEnv.createTemporaryView("sensor",sensorTable);
        Table resultSqlTable = tableEnv.sqlQuery("select id,ts,hashcode(id) from sensor");

        // 6.输出
        tableEnv.toAppendStream(resultTable,Row.class).print("table");
        tableEnv.toAppendStream(resultSqlTable,Row.class).print("sql");

        env.execute();
    }

    // 实现自定义hashCode类,需要是一个public类
    public static class HashCode extends ScalarFunction {

        private int factor = 13;

        public HashCode(int factor) {
            this.factor = factor;
        }

        // 函数名必须为eval，并且是public的
        public int eval(String str) {
            return str.hashCode() * factor;
        }
    }
}
